package com.wlyuan.index.domain.index.service.reindex;

import com.alibaba.fastjson.JSON;
import com.wlyuan.core.domain.valueobj.CombineId;
import com.wlyuan.index.api.dto.combine.CombineReindexAllDTO;
import com.wlyuan.index.api.dto.combine.CombineReindexDTO;
import com.wlyuan.index.domain.index.factory.CombineIndexFactory;
import com.wlyuan.index.domain.index.service.CombineIndexRenewService;
import com.wlyuan.index.domain.source.factory.SourceFactory;
import com.wlyuan.index.remote.facade.CombineFacade;
import com.wlyuan.index.remote.facade.QuotationFacade;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.StopWatch;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author yuanjie
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class CombineReindexService {
    private final SourceFactory sourceFactory;
    private final CombineIndexFactory indexFactory;
    private final CombineIndexRenewService renewService;
    private final CombineFacade combineFacade;
    private final QuotationFacade quotationFacade;
    private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(4,
            4, 5, TimeUnit.MINUTES, new LinkedBlockingQueue<>(),
            new CustomizableThreadFactory("combine-reindex-pool"));

    public void reindex(CombineId combineId) {
        try {
            var combine = combineFacade.getCombineById(combineId);
            if (null == combine) {
                logger.error("合单不存在: {}", combineId);
                return;
            }

            sourceFactory.create(combine);
            var quotations = quotationFacade.listQuotations(combineId);
            if (CollectionUtils.isNotEmpty(quotations)) {
                quotations.forEach(sourceFactory::create);
            }

            // 先删除索引
            indexFactory.remove(combineId);
            // 再创建索引
            renewService.add(combineId);
        } catch (Exception e) {
            logger.error("重建合单索引异常: {}", combineId, e);
        }
    }

    public void reindex(CombineReindexDTO reindex) {
        logger.info("重建合单索引: {}", JSON.toJSON(reindex));
        CompletableFuture.runAsync(() -> {
            var page = 1;
            var size = 1000;
            var totalSize = new AtomicLong(0);
            var completedSize = new AtomicLong(0L);
            var combineIds = combineFacade.listIds(reindex, page++, size);
            while (CollectionUtils.isNotEmpty(combineIds)) {
                logger.info("创建重建合单索引任务: {} {} {}", reindex, page - 1, size);
                this.reindexAsync(combineIds, completedSize);
                totalSize.addAndGet(combineIds.size());
                combineIds = combineFacade.listIds(reindex, page++, size);
            }
            logger.info("重建合单索引任务提交完成: {} {} {}", reindex, page - 2, totalSize.get());
        });
    }


    private void reindexAsync(List<CombineId> combineIds, AtomicLong completedSize) {
        CompletableFuture.runAsync(() -> {
            if (CollectionUtils.isEmpty(combineIds)) {
                return;
            }
            StopWatch watch = new StopWatch();
            watch.start();
            combineIds.forEach(combineId -> {
                try {
                    this.reindex(combineId);
                } catch (Exception e) {
                    logger.error("Reindex combine error: {}", combineId, e);
                }
            });
            watch.stop();
            completedSize.addAndGet(combineIds.size());
            logger.info("重建合单索引任务完成, 耗时: {}ms {} {}", watch.getTotalTimeMillis(),
                    combineIds.size(), completedSize.get());
        }, EXECUTOR);
    }

    public void reindexAll(CombineReindexAllDTO reindexAll) {
        CombineReindexDTO reindex = new CombineReindexDTO();
        BeanUtils.copyProperties(reindexAll, reindex);
        this.reindex(reindex);
    }

    public void reindexAll(List<CombineId> combineIds) {
        var completedSize = new AtomicLong(0L);
        this.reindexAsync(combineIds, completedSize);
    }
}
